/*-------------------------------------------------------------------------
 *
 * nodeGatherMerge.c
 *		在多个工作线程中扫描计划，并进行有序合并。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/executor/nodeGatherMerge.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/relscan.h"
#include "access/xact.h"
#include "executor/execdebug.h"
#include "executor/execParallel.h"
#include "executor/nodeGatherMerge.h"
#include "executor/nodeSubplan.h"
#include "executor/tqueue.h"
#include "lib/binaryheap.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "utils/memutils.h"
#include "utils/rel.h"

/*
 * 当我们从工作者读取元组时，最好尽可能一次读取多个元组以提高效率：
 * 这最小化了上下文切换的开销。
 * 但是一次读取太多则浪费内存而不会提高性能。
 * 我们最多将读取 MAX_TUPLE_STORE 个元组（除了第一个）。
 */
#define MAX_TUPLE_STORE 10

/*
 * 每个工作者的待处理元组数组。 这保存了我们能够从工作者获取的附加元组，
 * 但仍无法处理。此外，* 这个结构保存“完成”标志，指示工作者已知没有更多元组。 
 * （我们不为领导者使用这个结构；我们不保存领导者的任何待处理元组，而需要本地扫描的标志作为其“完成”指示符。）
 */
typedef struct GMReaderTupleBuffer
{
	MinimalTuple *tuple;		/* 长度为 MAX_TUPLE_STORE 的数组 */
	int			nTuples;		/* 当前存储的元组数量 */
	int			readCounter;	/* 下一个提取的元组索引 */
	bool		done;			/* 如果读取器已知耗尽则为真 */
} GMReaderTupleBuffer;

static TupleTableSlot *fc_ExecGatherMerge(PlanState *fc_pstate);
static int32 fc_heap_compare_slots(Datum fc_a, Datum fc_b, void *fc_arg);
static TupleTableSlot *fc_gather_merge_getnext(GatherMergeState *fc_gm_state);
static MinimalTuple fc_gm_readnext_tuple(GatherMergeState *fc_gm_state, int fc_nreader,
									  bool fc_nowait, bool *fc_done);
static void fc_ExecShutdownGatherMergeWorkers(GatherMergeState *fc_node);
static void fc_gather_merge_setup(GatherMergeState *fc_gm_state);
static void fc_gather_merge_init(GatherMergeState *fc_gm_state);
static void fc_gather_merge_clear_tuples(GatherMergeState *fc_gm_state);
static bool fc_gather_merge_readnext(GatherMergeState *fc_gm_state, int fc_reader,
								  bool fc_nowait);
static void fc_load_tuple_array(GatherMergeState *fc_gm_state, int fc_reader);

/* ----------------------------------------------------------------
 *		ExecInitGather
 * ----------------------------------------------------------------
 */
GatherMergeState *
ExecInitGatherMerge(GatherMerge *fc_node, EState *fc_estate, int fc_eflags)
{
	GatherMergeState *fc_gm_state;
	Plan	   *fc_outerNode;
	TupleDesc	fc_tupDesc;

	/* Gather 合并节点没有 innerPlan 节点。 */
	Assert(innerPlan(fc_node) == NULL);

	/*
	 * 创建状态结构
	 */
	fc_gm_state = makeNode(GatherMergeState);
	fc_gm_state->ps.plan = (Plan *) fc_node;
	fc_gm_state->ps.state = fc_estate;
	fc_gm_state->ps.ExecProcNode = fc_ExecGatherMerge;

	fc_gm_state->initialized = false;
	fc_gm_state->gm_initialized = false;
	fc_gm_state->tuples_needed = -1;

	/*
	 * 杂项初始化
	 *
	 * 为节点创建表达式上下文
	 */
	ExecAssignExprContext(fc_estate, &fc_gm_state->ps);

	/*
	 * GatherMerge 不支持检查 qual（在子节点中进行更高效）。
	 */
	Assert(!fc_node->plan.qual);

	/*
	 * 现在初始化外部计划
	 */
	fc_outerNode = outerPlan(fc_node);
	outerPlanState(fc_gm_state) = ExecInitNode(fc_outerNode, fc_estate, fc_eflags);

	/*
	 * 领导者可能直接访问 ExecProcNode 结果（如果需要本地扫描），
	 * 或通过元组队列从工作进程获取。因此，我们不能
	 * 简单地依赖于在此节点内评估的表达式的槽类型是固定的。
	 */
	fc_gm_state->ps.outeropsset = true;
	fc_gm_state->ps.outeropsfixed = false;

	/*
	 * 将元组描述符存储到 gather merge 状态中，以便我们可以在初始化 gather merge 插槽时使用它。
	 */
	fc_tupDesc = ExecGetResultType(outerPlanState(fc_gm_state));
	fc_gm_state->tupDesc = fc_tupDesc;

	/*
	 * 初始化结果类型和投影。
	 */
	ExecInitResultTypeTL(&fc_gm_state->ps);
	ExecConditionalAssignProjectionInfo(&fc_gm_state->ps, fc_tupDesc, OUTER_VAR);

	/*
	 * 在没有投影的情况下，结果槽类型并不是简单已知的，见上面的评论。
	 */
	if (fc_gm_state->ps.ps_ProjInfo == NULL)
	{
		fc_gm_state->ps.resultopsset = true;
		fc_gm_state->ps.resultopsfixed = false;
	}

	/*
	 * 初始化排序键信息
	 */
	if (fc_node->numCols)
	{
		int			fc_i;

		fc_gm_state->gm_nkeys = fc_node->numCols;
		fc_gm_state->gm_sortkeys =
			palloc0(sizeof(SortSupportData) * fc_node->numCols);

		for (fc_i = 0; fc_i < fc_node->numCols; fc_i++)
		{
			SortSupport fc_sortKey = fc_gm_state->gm_sortkeys + fc_i;

			fc_sortKey->ssup_cxt = CurrentMemoryContext;
			fc_sortKey->ssup_collation = fc_node->collations[fc_i];
			fc_sortKey->ssup_nulls_first = fc_node->nullsFirst[fc_i];
			fc_sortKey->ssup_attno = fc_node->sortColIdx[fc_i];

			/*
			 * 我们在这里不执行缩略键转换，原因与它在 MergeAppend 中未使用的原因相同
			 */
			fc_sortKey->abbreviate = false;

			PrepareSortSupportFromOrderingOp(fc_node->sortOperators[fc_i], fc_sortKey);
		}
	}

	/* 现在为 gather merge 分配工作空间 */
	fc_gather_merge_setup(fc_gm_state);

	return fc_gm_state;
}

/* ----------------------------------------------------------------
 *		ExecGatherMerge(node)
 *
 *		通过多个工作者扫描关系并返回
 *		下一个合格元组。
 * ----------------------------------------------------------------
 */
static TupleTableSlot * fc_ExecGatherMerge(PlanState *fc_pstate)
{
	GatherMergeState *fc_node = castNode(GatherMergeState, fc_pstate);
	TupleTableSlot *fc_slot;
	ExprContext *fc_econtext;

	CHECK_FOR_INTERRUPTS();

	/*
	 * 与 Gather 一样，我们只有在实际执行该节点时才会启动工作者。
	 */
	if (!fc_node->initialized)
	{
		EState	   *fc_estate = fc_node->ps.state;
		GatherMerge *fc_gm = castNode(GatherMerge, fc_node->ps.plan);

		/*
		 * 有时我们可能不得不在没有并行处理的情况下运行；但如果并行
		 * 模式处于活动状态，那么我们可以尝试启动一些工作进程。
		 */
		if (fc_gm->num_workers > 0 && fc_estate->es_use_parallel_mode)
		{
			ParallelContext *fc_pcxt;

			/* 初始化或重新初始化工作进程所需的共享状态。 */
			if (!fc_node->pei)
				fc_node->pei = ExecInitParallelPlan(fc_node->ps.lefttree,
												 fc_estate,
												 fc_gm->initParam,
												 fc_gm->num_workers,
												 fc_node->tuples_needed);
			else
				ExecParallelReinitialize(fc_node->ps.lefttree,
										 fc_node->pei,
										 fc_gm->initParam);

			/* 尝试启动工作者。 */
			fc_pcxt = fc_node->pei->pcxt;
			LaunchParallelWorkers(fc_pcxt);
			/* 我们保存启动的工作进程数量，以便 EXPLAIN 受益 */
			fc_node->nworkers_launched = fc_pcxt->nworkers_launched;

			/* 设置元组队列读取器以读取结果。 */
			if (fc_pcxt->nworkers_launched > 0)
			{
				ExecParallelCreateReaders(fc_node->pei);
				/* 制作一个工作数组，显示活动的读取器 */
				fc_node->nreaders = fc_pcxt->nworkers_launched;
				fc_node->reader = (TupleQueueReader **)
					palloc(fc_node->nreaders * sizeof(TupleQueueReader *));
				memcpy(fc_node->reader, fc_node->pei->reader,
					   fc_node->nreaders * sizeof(TupleQueueReader *));
			}
			else
			{
				/* 没有工作进程？那就算了。 */
				fc_node->nreaders = 0;
				fc_node->reader = NULL;
			}
		}

		/* 如果启用或没有选择，则允许领导者参与 */
		if (parallel_leader_participation || fc_node->nreaders == 0)
			fc_node->need_to_scan_locally = true;
		fc_node->initialized = true;
	}

	/*
	 * 重置每个元组的内存上下文，以释放在上一个元组周期中分配的任何表达式评估
	 * 存储。
	 */
	fc_econtext = fc_node->ps.ps_ExprContext;
	ResetExprContext(fc_econtext);

	/*
	 * 获取下一个元组，可能来自我们的一个工作进程，或者通过自己运行计划
	 * 来获取。
	 */
	fc_slot = fc_gather_merge_getnext(fc_node);
	if (TupIsNull(fc_slot))
		return NULL;

	/* 如果不需要投影，我们就完成了。 */
	if (fc_node->ps.ps_ProjInfo == NULL)
		return fc_slot;

	/*
	 * 使用 ExecProject() 形成结果元组，并返回它。
	 */
	fc_econtext->ecxt_outertuple = fc_slot;
	return ExecProject(fc_node->ps.ps_ProjInfo);
}

/* ----------------------------------------------------------------
 *		ExecEndGatherMerge
 *
 *		释放通过 C 例程分配的任何存储。
 * ----------------------------------------------------------------
 */
void ExecEndGatherMerge(GatherMergeState *fc_node)
{
	ExecEndNode(outerPlanState(fc_node));	/* 让子进程先清理 */
	ExecShutdownGatherMerge(fc_node);
	ExecFreeExprContext(&fc_node->ps);
	if (fc_node->ps.ps_ResultTupleSlot)
		ExecClearTuple(fc_node->ps.ps_ResultTupleSlot);
}

/* ----------------------------------------------------------------
 *		ExecShutdownGatherMerge
 *
 *		销毁并行工作者的设置，包括并行上下文。
 * ----------------------------------------------------------------
 */
void ExecShutdownGatherMerge(GatherMergeState *fc_node)
{
	fc_ExecShutdownGatherMergeWorkers(fc_node);

	/* 现在销毁并行上下文。 */
	if (fc_node->pei != NULL)
	{
		ExecParallelCleanup(fc_node->pei);
		fc_node->pei = NULL;
	}
}

/* ----------------------------------------------------------------
 *		ExecShutdownGatherMergeWorkers
 *
 *		停止所有并行工作者。
 * ----------------------------------------------------------------
 */
static void fc_ExecShutdownGatherMergeWorkers(GatherMergeState *fc_node)
{
	if (fc_node->pei != NULL)
		ExecParallelFinish(fc_node->pei);

	/* 刷新读者数组的本地副本 */
	if (fc_node->reader)
		pfree(fc_node->reader);
	fc_node->reader = NULL;
}

/* ----------------------------------------------------------------
 *		ExecReScanGatherMerge
 *
 *		准备重新扫描 GatherMerge 的结果。
 * ----------------------------------------------------------------
 */
void ExecReScanGatherMerge(GatherMergeState *fc_node)
{
	GatherMerge *fc_gm = (GatherMerge *) fc_node->ps.plan;
	PlanState  *fc_outerPlan = outerPlanState(fc_node);

	/* 确保任何现有工作者都已优雅地关闭 */
	fc_ExecShutdownGatherMergeWorkers(fc_node);

	/* 释放任何未使用的元组，以避免在重新扫描时内存泄漏 */
	fc_gather_merge_clear_tuples(fc_node);

	/* 标记节点，以便在下次调用时重建共享状态 */
	fc_node->initialized = false;
	fc_node->gm_initialized = false;

	/*
	 * 设置子节点的 chgParam，告诉它下一个扫描可能在领导进程中
	 * 交付不同的行集。 （整体行集不应该改变，但领导进程的子集可能会；因此在这里和并行表扫描节点之间的节点
	 * 不得基于不变的行集假设进行优化。）
	 */
	if (fc_gm->rescan_param >= 0)
		fc_outerPlan->chgParam = bms_add_member(fc_outerPlan->chgParam,
											 fc_gm->rescan_param);

	/*
	 * 如果子节点的 chgParam 不是 null，则计划将由第一个 ExecProcNode 重新扫描。
	 * 注意：因为如果我们有 rescan_param，这样做不会产生任何结果，所以当前保证并行感知的子节点
	 * 在收到 ReInitializeDSM 调用之前不会看到 ReScan 调用。
	 * 这种顺序可能不是可靠的。 一个好的经验法则是 ReInitializeDSM 应该仅重置共享状态，ReScan
	 * 应仅重置本地状态，且任何依赖于这两个步骤完成的内容必须等到第一次 ExecProcNode 调用后才能执行。
	 */
	if (fc_outerPlan->chgParam == NULL)
		ExecReScan(fc_outerPlan);
}

/*
 * 设置 Gather Merge 所需的数据结构。
 *
 * 我们根据 gm->num_workers 分配这些结构，这个值是我们实际会有的工作者的上限。在
 * 重新扫描期间，我们将结构重置为空。这种方法简化了在重新扫描时不泄漏内存的过程。
 *
 * 在 gm_slots[] 数组中，索引 0 用于领导者，索引 1 到 n 用于工作者。放入 gm_heap 的值
 * 对应于 gm_slots[] 中的索引。然而，gm_tuple_buffers[] 数组是从 0 到 n-1 索引的；
 * 它没有为领导者设立条目。
 */
static void fc_gather_merge_setup(GatherMergeState *fc_gm_state)
{
	GatherMerge *fc_gm = castNode(GatherMerge, fc_gm_state->ps.plan);
	int			fc_nreaders = fc_gm->num_workers;
	int			fc_i;

	/*
	 * 为工作者的数量分配 gm_slots，并为领导者再多分配一个槽位。
	 * 槽位 0 始终用于领导者。领导者总是调用 ExecProcNode() 来
	 * 读取元组，然后直接将其存储到其 gm_slots 条目中。
	 * 对于其他槽位，下面的代码将调用 ExecInitExtraTupleSlot() 来
	 * 为工作者的结果创建槽位。注意，在任何一次扫描中，我们可能会有
	 * 少于 num_workers 可用的工作者，在这种情况下，多余的数组条目将不会使用。
	 */
	fc_gm_state->gm_slots = (TupleTableSlot **)
		palloc0((fc_nreaders + 1) * sizeof(TupleTableSlot *));

	/* 为每个工作者分配元组槽和元组数组 */
	fc_gm_state->gm_tuple_buffers = (GMReaderTupleBuffer *)
		palloc0(fc_nreaders * sizeof(GMReaderTupleBuffer));

	for (fc_i = 0; fc_i < fc_nreaders; fc_i++)
	{
		/* 分配长度为 MAX_TUPLE_STORE 的元组数组 */
		fc_gm_state->gm_tuple_buffers[fc_i].tuple =
			(MinimalTuple *) palloc0(sizeof(MinimalTuple) * MAX_TUPLE_STORE);

		/* 初始化工作者的元组槽 */
		fc_gm_state->gm_slots[fc_i + 1] =
			ExecInitExtraTupleSlot(fc_gm_state->ps.state, fc_gm_state->tupDesc,
								   &TTSOpsMinimalTuple);
	}

	/* 为合并分配资源 */
	fc_gm_state->gm_heap = binaryheap_allocate(fc_nreaders + 1,
											fc_heap_compare_slots,
											fc_gm_state);
}

/*
 * 初始化 Gather Merge。
 *
 * 重置数据结构以确保它们是空的。然后从领导者和每个工作者中提取至少一个元组（或设置其“完成”指示符），并设置堆。
 */
static void fc_gather_merge_init(GatherMergeState *fc_gm_state)
{
	int			fc_nreaders = fc_gm_state->nreaders;
	bool		fc_nowait = true;
	int			fc_i;

	/* 断言 gather_merge_setup 释放了足够的空间 */
	Assert(fc_nreaders <= castNode(GatherMerge, fc_gm_state->ps.plan)->num_workers);

	/* 将领导者的元组槽重置为空 */
	fc_gm_state->gm_slots[0] = NULL;

	/* 重置每个工作者的元组槽和元组数组 */
	for (fc_i = 0; fc_i < fc_nreaders; fc_i++)
	{
		/* 将元组数组重置为空 */
		fc_gm_state->gm_tuple_buffers[fc_i].nTuples = 0;
		fc_gm_state->gm_tuple_buffers[fc_i].readCounter = 0;
		/* 将完成标志重置为未完成 */
		fc_gm_state->gm_tuple_buffers[fc_i].done = false;
		/* 确保输出槽为空 */
		ExecClearTuple(fc_gm_state->gm_slots[fc_i + 1]);
	}

	/* 将二进制堆重置为空 */
	binaryheap_reset(fc_gm_state->gm_heap);

	/*
	 * 首先，在不等待的模式下尝试从每个工作者（包括领导者）读取一个元组。
	 * 之后，如果并非所有工作者都能够生成一个元组（或“完成”指示），
	 * 那么重新从剩余的工作者中读取，这次使用等待模式。将所有活跃的读者（至少生成 
	 * 一个元组的）加入堆中。
	 */
reread:
	for (fc_i = 0; fc_i <= fc_nreaders; fc_i++)
	{
		CHECK_FOR_INTERRUPTS();

		/* 如果已知已完成，则跳过此来源 */
		if ((fc_i == 0) ? fc_gm_state->need_to_scan_locally :
			!fc_gm_state->gm_tuple_buffers[fc_i - 1].done)
		{
			if (TupIsNull(fc_gm_state->gm_slots[fc_i]))
			{
				/* 还没有元组，尝试获取一个 */
				if (fc_gather_merge_readnext(fc_gm_state, fc_i, fc_nowait))
					binaryheap_add_unordered(fc_gm_state->gm_heap,
											 Int32GetDatum(fc_i));
			}
			else
			{
				/*
				 * 我们已经从这个工作者那里得到了至少一个元组，
				 * 但不妨再看看它现在是否有更多的准备好的元组。
				 */
				fc_load_tuple_array(fc_gm_state, fc_i);
			}
		}
	}

	/* 不需要重新检查领导者，因为对于它来说不等待模式无关紧要 */
	for (fc_i = 1; fc_i <= fc_nreaders; fc_i++)
	{
		if (!fc_gm_state->gm_tuple_buffers[fc_i - 1].done &&
			TupIsNull(fc_gm_state->gm_slots[fc_i]))
		{
			fc_nowait = false;
			goto reread;
		}
	}

	/* 现在构建堆。 */
	binaryheap_build(fc_gm_state->gm_heap);

	fc_gm_state->gm_initialized = true;
}

/*
 * 清理每个 Gather Merge 输入的元组表槽和任何未使用的待处理元组。
 */
static void fc_gather_merge_clear_tuples(GatherMergeState *fc_gm_state)
{
	int			fc_i;

	for (fc_i = 0; fc_i < fc_gm_state->nreaders; fc_i++)
	{
		GMReaderTupleBuffer *fc_tuple_buffer = &fc_gm_state->gm_tuple_buffers[fc_i];

		while (fc_tuple_buffer->readCounter < fc_tuple_buffer->nTuples)
			pfree(fc_tuple_buffer->tuple[fc_tuple_buffer->readCounter++]);

		ExecClearTuple(fc_gm_state->gm_slots[fc_i + 1]);
	}
}

/*
 * 读取 Gather Merge 的下一个元组。
 *
 * 从堆中提取排序后的元组。
 */
static TupleTableSlot * fc_gather_merge_getnext(GatherMergeState *fc_gm_state)
{
	int			fc_i;

	if (!fc_gm_state->gm_initialized)
	{
		/*
		 * 第一次遍历：从每个参与者中提取第一个元组，并
		 * 建立堆。
		 */
		fc_gather_merge_init(fc_gm_state);
	}
	else
	{
		/*
		 * 否则，从上次返回的参与者中提取下一个元组，
		 * 并将该参与者的索引重新插入堆中，因为它现在可能与
		 * 堆中其他元素的比较方式不同。
		 */
		fc_i = DatumGetInt32(binaryheap_first(fc_gm_state->gm_heap));

		if (fc_gather_merge_readnext(fc_gm_state, fc_i, false))
			binaryheap_replace_first(fc_gm_state->gm_heap, Int32GetDatum(fc_i));
		else
		{
			/* 读取器已耗尽，将其移出堆 */
			(void) binaryheap_remove_first(fc_gm_state->gm_heap);
		}
	}

	if (binaryheap_empty(fc_gm_state->gm_heap))
	{
		/*所有队列都已耗尽，堆也是如此*/
		fc_gather_merge_clear_tuples(fc_gm_state);
		return NULL;
	}
	else
	{
		/*从拥有领先元组的任何参与者返回下一个元组*/
		fc_i = DatumGetInt32(binaryheap_first(fc_gm_state->gm_heap));
		return fc_gm_state->gm_slots[fc_i];
	}
}

/*读取给定读取器的元组（以非等待模式），并加载到其元组数组中，直到我们有MAX_TUPLE_STORE个元组或者必须阻塞为止。*/
static void fc_load_tuple_array(GatherMergeState *fc_gm_state, int fc_reader)
{
	GMReaderTupleBuffer *fc_tuple_buffer;
	int			fc_i;

	/*如果这是领导者，则不执行任何操作。*/
	if (fc_reader == 0)
		return;

	fc_tuple_buffer = &fc_gm_state->gm_tuple_buffers[fc_reader - 1];

	/*如果数组中没有任何内容，则将计数器重置为零。*/
	if (fc_tuple_buffer->nTuples == fc_tuple_buffer->readCounter)
		fc_tuple_buffer->nTuples = fc_tuple_buffer->readCounter = 0;

	/*尝试填充数组中的额外槽位。*/
	for (fc_i = fc_tuple_buffer->nTuples; fc_i < MAX_TUPLE_STORE; fc_i++)
	{
		MinimalTuple fc_tuple;

		fc_tuple = fc_gm_readnext_tuple(fc_gm_state,
								  fc_reader,
								  true,
								  &fc_tuple_buffer->done);
		if (!fc_tuple)
			break;
		fc_tuple_buffer->tuple[fc_i] = fc_tuple;
		fc_tuple_buffer->nTuples++;
	}
}

/*将给定读取器的下一个元组存储到适当的槽中。返回成功时为真，失败时为假（要么读取器已耗尽，要么我们不想等待一个元组）。如果发现读取器已耗尽，则设置完成标志。*/
static bool fc_gather_merge_readnext(GatherMergeState *fc_gm_state, int fc_reader, bool fc_nowait)
{
	GMReaderTupleBuffer *fc_tuple_buffer;
	MinimalTuple fc_tup;

	/*如果我们被要求从领导者生成元组，那么我们只需像往常一样调用ExecProcNode来生成一个。*/
	if (fc_reader == 0)
	{
		if (fc_gm_state->need_to_scan_locally)
		{
			PlanState  *fc_outerPlan = outerPlanState(fc_gm_state);
			TupleTableSlot *fc_outerTupleSlot;
			EState	   *fc_estate = fc_gm_state->ps.state;

			/* 在执行计划时安装我们的 DSA 区域。 */
			fc_estate->es_query_dsa = fc_gm_state->pei ? fc_gm_state->pei->area : NULL;
			fc_outerTupleSlot = ExecProcNode(fc_outerPlan);
			fc_estate->es_query_dsa = NULL;

			if (!TupIsNull(fc_outerTupleSlot))
			{
				fc_gm_state->gm_slots[0] = fc_outerTupleSlot;
				return true;
			}
			/*need_to_scan_locally作为领导者的“完成”标志*/
			fc_gm_state->need_to_scan_locally = false;
		}
		return false;
	}

	/*否则，检查相关元组缓冲区的状态。*/
	fc_tuple_buffer = &fc_gm_state->gm_tuple_buffers[fc_reader - 1];

	if (fc_tuple_buffer->nTuples > fc_tuple_buffer->readCounter)
	{
		/*返回任何先前读取且仍然缓冲的元组。*/
		fc_tup = fc_tuple_buffer->tuple[fc_tuple_buffer->readCounter++];
	}
	else if (fc_tuple_buffer->done)
	{
		/*读取器已知耗尽。*/
		return false;
	}
	else
	{
		/*读取并缓冲下一个元组。*/
		fc_tup = fc_gm_readnext_tuple(fc_gm_state,
								fc_reader,
								fc_nowait,
								&fc_tuple_buffer->done);
		if (!fc_tup)
			return false;

		/*尝试在非等待模式下读取更多元组，并将它们存储在读取器的待处理元组数组中。*/
		fc_load_tuple_array(fc_gm_state, fc_reader);
	}

	Assert(fc_tup);

	/*为给定元组构建TupleTableSlot*/
	ExecStoreMinimalTuple(fc_tup,	/* 存储的元组 */
						  fc_gm_state->gm_slots[fc_reader],	/*存储元组的槽*/
						  true);	/*使用后释放元组*/

	return true;
}

/*尝试从给定的工作者读取元组。*/
static MinimalTuple fc_gm_readnext_tuple(GatherMergeState *fc_gm_state, int fc_nreader, bool fc_nowait,
				  bool *fc_done)
{
	TupleQueueReader *fc_reader;
	MinimalTuple fc_tup;

	/* 检查异步事件，特别是来自工作进程的消息。 */
	CHECK_FOR_INTERRUPTS();

	/*尝试读取一个元组。请注意，TupleQueueReaderNext将仅对未能初始化的工作者返回NULL。我们将视该工作者为未生成任何元组；WaitForParallelWorkersToFinish将在我们到达那里时出错。*/
	fc_reader = fc_gm_state->reader[fc_nreader - 1];
	fc_tup = TupleQueueReaderNext(fc_reader, fc_nowait, fc_done);

	/*由于我们将在多个调用中缓冲这些元组，因此需要进行复制。*/
	return fc_tup ? heap_copy_minimal_tuple(fc_tup) : NULL;
}

/*我们为堆数组中的每个项目拥有一个槽。我们使用SlotNumber来存储槽索引。这实际上并不提供任何正式的类型安全，但使代码更具自文档化。*/
typedef int32 SlotNumber;

/*比较两个给定槽中的元组。*/
static int32 fc_heap_compare_slots(Datum fc_a, Datum fc_b, void *fc_arg)
{
	GatherMergeState *fc_node = (GatherMergeState *) fc_arg;
	SlotNumber	fc_slot1 = DatumGetInt32(fc_a);
	SlotNumber	fc_slot2 = DatumGetInt32(fc_b);

	TupleTableSlot *fc_s1 = fc_node->gm_slots[fc_slot1];
	TupleTableSlot *fc_s2 = fc_node->gm_slots[fc_slot2];
	int			fc_nkey;

	Assert(!TupIsNull(fc_s1));
	Assert(!TupIsNull(fc_s2));

	for (fc_nkey = 0; fc_nkey < fc_node->gm_nkeys; fc_nkey++)
	{
		SortSupport fc_sortKey = fc_node->gm_sortkeys + fc_nkey;
		AttrNumber	fc_attno = fc_sortKey->ssup_attno;
		Datum		fc_datum1,
					fc_datum2;
		bool		fc_isNull1,
					fc_isNull2;
		int			fc_compare;

		fc_datum1 = slot_getattr(fc_s1, fc_attno, &fc_isNull1);
		fc_datum2 = slot_getattr(fc_s2, fc_attno, &fc_isNull2);

		fc_compare = ApplySortComparator(fc_datum1, fc_isNull1,
									  fc_datum2, fc_isNull2,
									  fc_sortKey);
		if (fc_compare != 0)
		{
			INVERT_COMPARE_RESULT(fc_compare);
			return fc_compare;
		}
	}
	return 0;
}
